/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.client.consumer.rebalance;

import java.util.ArrayList;
import java.util.List;
import org.apache.rocketmq.client.consumer.AllocateMessageQueueStrategy;
import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.common.message.MessageQueue;

/**
 * Average Hashing queue algorithm


 消费端在消费消息时，并不会处理所有的队列里的数据，而是从可读取的队列中，按一定的算法选择队列进行消费。

 RocketMQ 提供了6种实现，默认为

 AllocateMachineRoomNearby

 AllocateMessageQueueAveragely

 AllocateMessageQueueAveragelyByCircle

 AllocateMessageQueueByConfig

 AllocateMessageQueueByMachineRoom

 AllocateMessageQueueConsistentHash

 默认情况下，为AllocateMessageQueueAveragely

 平均算法： 算出平均值，将连续的队列按平均值分配给每个消费者。如果队列数量可以整除消费数量，则平均分配，如果不能整除，则取余数，余数之前顺序的消费端，可以拿到平均值+1的数据
 平均算法顾名思义就是取平均值，该方法四个参数，consumerGroup（消费者组名称）、
 currentCID（当前消费者的id）、mqAll（当前topic下面所有的消息队列）、cidAll（当前消费者组下面所有的消费者id）。算法思想就是把算出平均值然后将连续的队列分配给每个消费者。假设队列大小是8（编号0-7），消费者数量3（编号0-2），分配结果就是：
 消费者0：队列0，1，2；
 消费者1：队列3，4，5；
 消费者2：队列6，7。

 */
public class AllocateMessageQueueAveragely implements AllocateMessageQueueStrategy {
    private final InternalLogger log = ClientLogger.getLog();

    @Override
    public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll,
        List<String> cidAll) {
        if (currentCID == null || currentCID.length() < 1) {
            throw new IllegalArgumentException("currentCID is empty");
        }
        if (mqAll == null || mqAll.isEmpty()) {
            throw new IllegalArgumentException("mqAll is null or mqAll empty");
        }
        if (cidAll == null || cidAll.isEmpty()) {
            throw new IllegalArgumentException("cidAll is null or cidAll empty");
        }

        List<MessageQueue> result = new ArrayList<MessageQueue>();
        if (!cidAll.contains(currentCID)) {
            log.info("[BUG] ConsumerGroup: {} The consumerId: {} not in cidAll: {}",
                consumerGroup,
                currentCID,
                cidAll);
            return result;
        }

        int index = cidAll.indexOf(currentCID);
        int mod = mqAll.size() % cidAll.size();
        int averageSize =
            mqAll.size() <= cidAll.size() ? 1 : (mod > 0 && index < mod ? mqAll.size() / cidAll.size()
                + 1 : mqAll.size() / cidAll.size());
        int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;
        int range = Math.min(averageSize, mqAll.size() - startIndex);
        for (int i = 0; i < range; i++) {
            result.add(mqAll.get((startIndex + i) % mqAll.size()));
        }
        return result;
    }

    @Override
    public String getName() {
        return "AVG";
    }
}
